package cn.tecnova.cd

import cn.tecnova.utils.ConfigHandler
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{Dataset, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
import org.elasticsearch.spark.rdd.EsSpark

/**
  * description:
  * 717 6126
  **/
object TecnovaPartners {
  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      //      .setMaster("local[*]")
      .set("es.index.auto.create", "true")
      .set("es.nodes", ConfigHandler.esNodes)
      .set("es.port", ConfigHandler.esPort)
      .set("es.nodes.wan.only", "true")

    val sc = new SparkContext(conf)
    val sqLContext = new SQLContext(sc)

    import sqLContext.implicits._
    val rowDF = sqLContext.read.jdbc("jdbc:mysql://www.slave4.com:3306/tec_partners", "qichacha_partners","id",1,7176126,10, ConfigHandler.props2)

    //导入写es相关的包
    import org.elasticsearch.spark.sql._

    val result: Dataset[TecnovaPartners] = rowDF.map(row => {
      TecnovaPartners(
        row.getAs[Int]("id"),
        row.getAs[String]("KeyNo"),
        row.getAs[Int]("Org"),
        row.getAs[Int]("SeqId"),
        row.getAs[String]("CompanyId"),
        row.getAs[String]("StockName"),
        row.getAs[String]("StockType"),
        row.getAs[String]("StockPercent"),
        row.getAs[String]("IdentifyType"),
        row.getAs[String]("IdentifyNo"),
        row.getAs[String]("ShouldCapi"),
        row.getAs[String]("ShoudDate"),
        row.getAs[String]("InvestType"),
        row.getAs[String]("InvestName"),
        row.getAs[String]("RealCapi"),
        row.getAs[String]("CapiDate"),
        row.getAs[String]("Address"),
        row.getAs[String]("CapiRate"),
        row.getAs[String]("LargePartner")

      )
    })
//    result.saveToEs("tecnova_partners2" + "/company_partners")

    EsSpark.saveToEs(result.rdd,"tecnova_partners/company_partners",Map("es.mapping.id" -> "id"))


    sc.stop()

  }

}

case class TecnovaPartners(

                         var id:Long,
                         var keyNo:String,
                         var org:Int,
                         var seqId:Int,
                         var companyId:String,
                         var stockName:String,
                         var stockType:String,
                         var stockPercent:String,
                         var identifyType:String,
                         var identifyNo:String,
                         var shouldCapi:String,
                         var shoudDate:String,
                         var investType:String,
                         var investName:String,
                         var realCapi:String,
                         var capiDate:String,
                         var address:String,
                         var capiRate:String,
                         var largePartner:String

                          )
